/*
 * [y] hybris Platform
 *
 * Copyright (c) 2000-2017 SAP SE
 * All rights reserved.
 *
 * This software is the confidential and proprietary information of SAP
 * Hybris ("Confidential Information"). You shall not disclose such
 * Confidential Information and shall use it only in accordance with the
 * terms of the license agreement you entered into with SAP Hybris.
 */
package acc.asset.sync.creator;

import de.hybris.deltadetection.ChangeDetectionService;
import de.hybris.platform.core.model.media.MediaFolderModel;
import de.hybris.platform.core.model.media.MediaModel;
import de.hybris.platform.servicelayer.cronjob.CronJobService;
import de.hybris.platform.servicelayer.event.EventService;
import de.hybris.platform.servicelayer.media.MediaService;
import de.hybris.platform.servicelayer.model.ModelService;
import de.hybris.platform.servicelayer.search.FlexibleSearchService;
import de.hybris.platform.task.TaskService;
import de.hybris.platform.util.Config;
import de.hybris.y2ysync.model.media.SyncImpExMediaModel;
import de.hybris.y2ysync.task.dao.Y2YSyncDAO;
import de.hybris.y2ysync.task.internal.SyncTaskFactory;
import de.hybris.y2ysync.task.runner.Y2YSyncContext;
import de.hybris.y2ysync.task.runner.internal.DataHubRequestCreator;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Properties;

import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.log4j.Logger;
import org.springframework.security.oauth2.common.util.SerializationUtils;
import org.springframework.web.client.RestTemplate;

import acc.asset.sync.bean.DemoSyncData;
import acc.asset.sync.bean.DemoSyncRequest;


/**
 *
 */
public class DemoDataHubRequestCreator extends DataHubRequestCreator
{
	private static final Logger LOG = Logger.getLogger(DemoDataHubRequestCreator.class);

	private Y2YSyncDAO y2YSyncDAO;
	private RestTemplate restTemplate;
	private EventService eventService;
	private ModelService modelService;

	private MediaService mediaService;
	private SyncTaskFactory syncTaskFactory;
	private TaskService taskService;

	private CronJobService cronJobService;

	private ChangeDetectionService changeDetectionService;

	private FlexibleSearchService flexibleSearchService;

	@Override
	public void sendRequest(final Y2YSyncContext syncContext)
	{
		Objects.requireNonNull(syncContext, "syncContext is required");
		final String syncExecutionId = syncContext.getSyncExecutionId();
		final List<SyncImpExMediaModel> list = getY2YSyncDAO().findSyncMediasBySyncCronJob(syncExecutionId);

		onEvent(syncExecutionId, list);
	}

	protected void onEvent(final String syncExecutionId, final List<SyncImpExMediaModel> mediaList)
	{
		LOG.info(String.format("syncExecutionID:%s", syncExecutionId));

		if (mediaList != null)
		{
			if (mediaList.isEmpty())
			{
				return;
			}

			final Producer<String, byte[]> producer = new KafkaProducer<>(createProducerProperties());

			final Callback producerCallback = (final RecordMetadata metadata, final Exception e) -> {
				if (e != null)
				{
					LOG.error(e.getMessage(), e);
				}
				else
				{
					LOG.info("The offset of the record we just send is:" + metadata);
				}
			};

			try
			{
				final String homeURL = getHomeUrl();
				final DemoSyncRequest demoSyncRequest = new DemoSyncRequest();
				demoSyncRequest.setSourcePlatformUrl(homeURL);
				demoSyncRequest.setSyncExecutionId(syncExecutionId);
				final List<DemoSyncData> syncDataList = createSyncDataList(mediaList, homeURL);
				if (syncDataList == null || syncDataList.isEmpty())
				{
					LOG.info("syncDataList is empty");
					return;
				}

				demoSyncRequest.setDemoSyncDataList(syncDataList);

				final ProducerRecord<String, byte[]> record = new ProducerRecord<String, byte[]>(getTopic(),
						String.valueOf(System.currentTimeMillis()), SerializationUtils.serialize(demoSyncRequest));

				producer.send(record, producerCallback);
				producer.flush();
			}
			catch (final Exception e)
			{
				LOG.info(e.getMessage(), e);
			}
			finally
			{
				producer.close();
			}

		}

	}//end method

	private Properties createProducerProperties()
	{
		final Properties props = new Properties();
		props.put("bootstrap.servers", getKafkaBootstrapServers());
		props.put("acks", Config.getString("kafka.producer.acks", "all"));
		props.put("retries", Integer.valueOf(Config.getInt("kafka.producer.retries", 0)));
		props.put("batch.size", Integer.valueOf(Config.getInt("kafka.producer.batch.size", 16384)));
		props.put("linger.ms", Integer.valueOf(Config.getInt("kafka.producer.linger.ms", 1)));
		props.put("buffer.memory", Integer.valueOf(Config.getInt("kafka.producer.buffer.memory", 33554432)));
		props.put("max.request.size", Integer.valueOf(Config.getInt("kafka.producer.max.request.size", 1048576000)));

		final String keySerizlizer = Config.getString("kafka.producer.key.serializer", "");
		final String valueSerizlizer = Config.getString("kafka.producer.value.serializer", "");
		if (StringUtils.isBlank(keySerizlizer) || StringUtils.isBlank(valueSerizlizer))
		{
			throw new IllegalArgumentException(
					"the value for kafka.producer.key.serializer or kafka.producer.value.serializer can not be empty");
		}
		props.put("key.serializer", keySerizlizer);
		props.put("value.serializer", valueSerizlizer);

		return props;

	}

	private String getTopic()
	{
		final String topic = Config.getString("kafka.producer.topic", "");
		if (StringUtils.isBlank(topic))
		{
			throw new IllegalArgumentException("the value for kafka.bootstrap.servers can not be empty");
		}

		return topic;
	}

	/**
	 *
	 */
	private List<DemoSyncData> createSyncDataList(final List<SyncImpExMediaModel> mediaList, final String homeURL)
			throws IOException
	{
		LOG.info("List<SyncImpExMediaModel>.size:" + mediaList.size());
		final List<DemoSyncData> syncDataList = new ArrayList<>();
		for (final SyncImpExMediaModel media : mediaList)
		{
			final MediaModel impexMedia = media.getMediaArchive();
			if (impexMedia != null)
			{
				LOG.info("media code:" + impexMedia.getCode());
				LOG.info("media pk:" + impexMedia.getPk());
			}
			LOG.info("SyncImpExMedia syncType:" + media.getSyncType());

			/////////////////////////
			final InputStream inputStream = getMediaService().getStreamFromMedia(media);
			final BufferedReader br = new BufferedReader(new InputStreamReader(inputStream));

			String string = null;
			final StringBuilder builder = new StringBuilder();
			try
			{
				while ((string = br.readLine()) != null)
				{
					if (!(string.indexOf("#") == 0))
					{
						builder.append(";").append(string).append("\n");
					}
				}
			}
			catch (final IOException e1)
			{
				e1.printStackTrace();

				return null;
			}
			finally
			{
				br.close();
			}

			//////////////
			syncDataList.add(new DemoSyncData(media.getImpexHeader(), builder.toString(), buildMediaUrl(media, homeURL),
					media.getCode(), media.getSyncType().getCode()));
		} //end
		return syncDataList;
	}

	private MediaFolderModel folderWithFallbackToRoot(final MediaFolderModel folder)
	{
		return folder == null ? this.getMediaService().getRootFolder() : folder;
	}

	private String buildMediaUrl(final MediaModel media, final String homeURL)
	{
		return String.valueOf(homeURL) + media.getURL();
	}

	public String getKafkaBootstrapServers()
	{
		final String servers = Config.getString("kafka.bootstrap.servers", "");
		if (StringUtils.isBlank(servers))
		{
			throw new IllegalArgumentException("the value for kafka.bootstrap.servers can not be empty");
		}

		return servers;
	}

	public String getY2YSyncWebRoot()
	{
		final String homeUrl = Config.getString("y2ysync.home.url", "http://localhost:9001");
		final String webRoot = Config.getString("y2ysync.webroot", "/y2ysync");
		return homeUrl + webRoot;
	}

	public String getHomeUrl()
	{
		return Config.getParameter("y2ysync.home.url");
	}

	/**
	 * @return the y2YSyncDAO
	 */
	public Y2YSyncDAO getY2YSyncDAO()
	{
		return y2YSyncDAO;
	}

	/**
	 * @param y2ySyncDAO
	 *           the y2YSyncDAO to set
	 */
	@Override
	public void setY2YSyncDAO(final Y2YSyncDAO y2ySyncDAO)
	{
		super.setY2YSyncDAO(y2ySyncDAO);
		y2YSyncDAO = y2ySyncDAO;
	}

	/**
	 * @return the restTemplate
	 */
	public RestTemplate getRestTemplate()
	{
		return restTemplate;
	}

	/**
	 * @param restTemplate
	 *           the restTemplate to set
	 */
	@Override
	public void setRestTemplate(final RestTemplate restTemplate)
	{
		super.setRestTemplate(restTemplate);
		this.restTemplate = restTemplate;
	}


	/**
	 * @return the eventService
	 */
	public EventService getEventService()
	{
		return eventService;
	}

	/**
	 * @param eventService
	 *           the eventService to set
	 */
	public void setEventService(final EventService eventService)
	{
		this.eventService = eventService;
	}

	/**
	 * @return the modelService
	 */
	public ModelService getModelService()
	{
		return modelService;
	}

	/**
	 * @return the mediaService
	 */
	public MediaService getMediaService()
	{
		return mediaService;
	}

	/**
	 * @return the taskService
	 */
	public TaskService getTaskService()
	{
		return taskService;
	}

	/**
	 * @param taskService
	 *           the taskService to set
	 */
	public void setTaskService(final TaskService taskService)
	{
		this.taskService = taskService;
	}

	/**
	 * @param mediaService
	 *           the mediaService to set
	 */
	public void setMediaService(final MediaService mediaService)
	{
		this.mediaService = mediaService;
	}

	/**
	 * @return the syncTaskFactory
	 */
	public SyncTaskFactory getSyncTaskFactory()
	{
		return syncTaskFactory;
	}

	/**
	 * @param syncTaskFactory
	 *           the syncTaskFactory to set
	 */
	public void setSyncTaskFactory(final SyncTaskFactory syncTaskFactory)
	{
		this.syncTaskFactory = syncTaskFactory;
	}

	/**
	 * @param modelService
	 *           the modelService to set
	 */
	public void setModelService(final ModelService modelService)
	{
		this.modelService = modelService;
	}

	/**
	 * @return the y2ySyncDAO
	 */
	public Y2YSyncDAO getY2ySyncDAO()
	{
		return y2YSyncDAO;
	}

	/**
	 * @param y2ySyncDAO
	 *           the y2ySyncDAO to set
	 */
	public void setY2ySyncDAO(final Y2YSyncDAO y2ySyncDAO)
	{
		this.y2YSyncDAO = y2ySyncDAO;
	}

	/**
	 * @return the changeDetectionService
	 */
	public ChangeDetectionService getChangeDetectionService()
	{
		return changeDetectionService;
	}

	/**
	 * @param changeDetectionService
	 *           the changeDetectionService to set
	 */
	public void setChangeDetectionService(final ChangeDetectionService changeDetectionService)
	{
		this.changeDetectionService = changeDetectionService;
	}

	/**
	 * @return the cronJobService
	 */
	public CronJobService getCronJobService()
	{
		return cronJobService;
	}

	/**
	 * @param cronJobService
	 *           the cronJobService to set
	 */
	public void setCronJobService(final CronJobService cronJobService)
	{
		this.cronJobService = cronJobService;
	}

	/**
	 * @return the flexibleSearchService
	 */
	public FlexibleSearchService getFlexibleSearchService()
	{
		return flexibleSearchService;
	}

	/**
	 * @param flexibleSearchService
	 *           the flexibleSearchService to set
	 */
	public void setFlexibleSearchService(final FlexibleSearchService flexibleSearchService)
	{
		this.flexibleSearchService = flexibleSearchService;
	}


}
